package com.aispeech.asr.apis.ssc.websocket.emitter;

import java.io.IOException;

import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.web.reactive.socket.WebSocketSession;

import com.aispeech.asr.apis.ssc.config.AppConfig;
import com.aispeech.asr.apis.ssc.data.domain.VadResult;
import com.aispeech.asr.apis.ssc.websocket.connection.DateBufferFactory;
import com.aispeech.asr.apis.ssc.websocket.connection.EmitterManager;
import com.aispeech.jarpc.ArpcException;
import com.aispeech.jarpc.JArpc;
import com.aispeech.jarpc.TextBin;
import com.aispeech.jarpc.message.ArpcMessage;
import com.aispeech.jarpc.op.sd.SdResult;
import com.aispeech.jarpc.op.vad.OfflineResult;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import lombok.ToString;
import lombok.extern.slf4j.Slf4j;

@ToString
@Slf4j
public class ClientEmitter extends Emitter {
	
	private String audioPath;
    private String requestQuery;
    private JSONObject startObj;
    private JSONObject audioObj;
    private AppConfig appConfig;
    private String sessionId;
    private JArpc jarpcVad;
    private JArpc jarpcSd;
    
    private long currentSpeechStartOffset;
    private long currentSpeechEndOffset;
    
    private int receiveTimes;
    private DataBuffer receiveBuffer; 
    
    private ByteBuf opBuffer;
    
    private boolean running = false; 
    
    public synchronized void writeReceiveBuffer(DataBuffer dataBuffer){
		if(receiveBuffer == null) {
			receiveBuffer = DateBufferFactory.getInstance().allocateBuffer();
		}
		receiveBuffer.write(dataBuffer);
    }
    
    public synchronized void writeOpBuffer(byte[] bytes) throws IOException {
    	if(opBuffer == null) {
    		opBuffer = Unpooled.buffer();
    	}
    	opBuffer.writeBytes(bytes);
    }

    public ClientEmitter(WebSocketSession session, String id, EmitterManager manager) {
        super(session, id, EmitterType.CLIENT, manager);
 
        running = true;
        //new Thread(this::integrationLoop).start();
    }
    
    
    public void setAudioPath(String audioPath) {
        this.audioPath = audioPath;
    }
    
    public String getAudioPath() {
        return audioPath;
    }
    
    public void setRequestQuery(String requestQuery) {
        this.requestQuery = requestQuery;
    }
    
    public String getRequestQuery() {
        return requestQuery;
    }
    
    public void setStartObj(JSONObject startObj) {
        this.startObj = startObj;
    }
    
    public JSONObject getStartObj() {
        return startObj;
    }
    
    public void setAudioObj(JSONObject audioObj) {
        this.audioObj = audioObj;
    }
    
    public JSONObject getAudioObj() {
        return audioObj;
    }
    
    public void setAppConfig(AppConfig appConfig) {
        this.appConfig = appConfig;
    }
    
    public AppConfig getAppConfig() {
        return appConfig;
    }
    
    public void setJarpcVad(JArpc jarpcVad) {
        this.jarpcVad = jarpcVad;
    }
    
    public JArpc getJarpcVad() {
        return jarpcVad;
    }
    
    public void setJarpcSd(JArpc jarpcSd) {
        this.jarpcSd = jarpcSd;
    }
    
    public JArpc getJarpcSd() {
        return jarpcSd;
    }
    
    public void setSessionId(String sesssionId) {
        this.sessionId = sesssionId;
    }
    
    public String getSessionId() {
        return sessionId;
    }
    
    public void setRunning(boolean flag) {
        this.running = flag;
    }
   
    public void addReceiveTimes() {
         receiveTimes++;
    }
    
    public int getReceiveTimes() {
        return receiveTimes;
    }
    
    public DataBuffer getReceiveBuffer() {
    	return this.receiveBuffer;
    }
    
    public void doProcess(DataBuffer dataBuffer) {
    	try {
    		//log.info("[doProcess] dataBuffer.readableByteCount():{}",dataBuffer.readableByteCount());
    		if(dataBuffer.readableByteCount() > 0) {
    			jarpcVad.feed(dataBuffer.asByteBuffer().array(), dataBuffer.readableByteCount());
    		}else {
    			ArpcMessage message = jarpcVad.stop(3);
    			TextBin textBin = TextBin.from(message.getBody());
    			doSpeakerDiarisation(textBin,1);
    		}
		} catch (IOException e) {
			log.error("[ClientEmitter IOException] doProcess failed.sessionId:{}",getId(),e);
			getManager().doErrorResponse(getId(), e);
		} catch (ArpcException e) {
			log.error("[ClientEmitter ArpcException] doProcess failed.sessionId:{}",getId(),e);
			getManager().doErrorResponse(getId(), e);
		}
    }
    
    public void doSpeakerDiarisation(TextBin textBin,int eof) {
    	try {
			if(textBin.getBin() != null && textBin.getBin().length > 0) {
				//log.info("[doSpeakerDiarisation] currentSpeechStartOffset:{},textBin.getBin().length:{}",currentSpeechStartOffset,textBin.getBin().length);
				long offset = currentSpeechStartOffset+textBin.getBin().length;
				//int startTime = AppConfig.pcmBytesToMsecs(currentSpeechStartOffset,audioObj.getIntValue("sampleRate"),audioObj.getIntValue("channel"),audioObj.getIntValue("sampleBytes"));
				//int endTime = AppConfig.pcmBytesToMsecs(offset,audioObj.getIntValue("sampleRate"),audioObj.getIntValue("channel"),audioObj.getIntValue("sampleBytes"));;
				String jsonStr = "{\"audio_info\":{\"state\":\"speech\",\"start\": %d, \"end\": %d}}";
				TextBin sdTextBin = new TextBin(String.format(jsonStr, currentSpeechStartOffset, offset), textBin.getBin());
				//log.info("[doSpeakerDiarisation] sdTextBin.getParam:{}",sdTextBin.getParam());
				byte[] bin = sdTextBin.asBuffer().array();
    			jarpcSd.feed(bin, bin.length);
    			currentSpeechStartOffset = offset;
    			//writeOpBuffer(textBin.getBin());
    		}else {
    			log.info("[doSpeakerDiarisation] =======textBin.getParam():{} ",textBin.getParam());
    			VadResult vadResult = JSON.parseObject(textBin.getParam(), VadResult.class);
    			if(vadResult.getStateOffsetList() != null && vadResult.size()>0) {
    				VadState vadState = vadResult.get(0);
    				//log.info("[doSpeakerDiarisation] =======vadState:{} ,before currentSpeechStartOffset:{}",vadState.getState(),currentSpeechStartOffset);
        			if(OfflineResult.State.SPEECH == vadState.getState()) {
        				currentSpeechStartOffset = vadState.getOffset();
        			}else {
        				long silOffset = vadState.getOffset();
        				String jsonStr = "{\"audio_info\":{\"state\":\"silence\",\"start\": %d, \"end\": %d}}";
        				TextBin sdTextBin = new TextBin(String.format(jsonStr, silOffset, silOffset), textBin.getBin());
        				//log.info("[doSpeakerDiarisation] sdTextBin.getParam:{}",sdTextBin.getParam());
        				byte[] bin = sdTextBin.asBuffer().array();
            			jarpcSd.feed(bin, bin.length);
            			//writeOpBuffer(textBin.getBin());
        				currentSpeechEndOffset = vadState.getOffset();
        			}
        			//log.info("[doSpeakerDiarisation] =======vadState:{} ,after currentSpeechStartOffset:{}",vadState.getState(),currentSpeechStartOffset);
    			}
    		}
			if(eof == 1) {
    			ArpcMessage message = jarpcSd.stop(3);
    			//log.info("[doSpeakerDiarisation] stop =======message:{},message.getBody:{} ",message,message.getBody());
    			
    			String arpcMessage = new String(message.getBody(),"utf-8");
    			log.info("[doSpeakerDiarisation] stop =======arpcMessage:{} ",arpcMessage);
    			
    			SdResult sdResult = JSON.parseObject(message.getBody(), SdResult.class);
    			getManager().doResponse(getId(), sdResult, 1);
    			
//    			if(opBuffer != null &&  opBuffer.readableBytes() > 0) {
//        			File receiveAudio = new File("/logs/audios/receive_"+sessionId+".wav");
//        			if(!receiveAudio.exists()) {
//        				receiveAudio.getParentFile().mkdirs();
//            			receiveAudio.createNewFile();
//            		}
////        			byte[] aaa= new byte[44];
////        			clientEmitter.getReceiveBuffer().read(aaa);
////        			log.info("=======clientEmitter.getReceiveBuffer():{}",Arrays.toString(aaa));
//        			
//        			InputStream  is = new ByteArrayInputStream(opBuffer.array());
//            		FileOutputStream fos = new FileOutputStream(receiveAudio);
//            		byte[] b = new byte[1024];
//            		int length;
//            		while((length = is.read(b))>0){
//            			fos.write(b,0,length);
//            		}
//            		is.close();
//            		fos.close();
//        		}
    			
    			
    		}
		} catch (IOException e) {
			log.error("[ClientEmitter IOException] doSpeakerDiarisation failed.sessionId:{}",getId(),e);
			getManager().doErrorResponse(getId(), e);
		} catch (ArpcException e) {
			log.error("[ClientEmitter ArpcException] doSpeakerDiarisation failed.sessionId:{}",getId(),e);
			getManager().doErrorResponse(getId(), e);
		}
    }
    
}
